package com.yutel.comp.queue.jms;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import com.yutel.comp.queue.QueueConfig;
import com.yutel.comp.queue.QueueConsumer;
import com.yutel.comp.queue.QueueLog;
import com.yutel.comp.queue.ReceiveListener;

public class JmsConsumer extends Thread implements QueueConsumer {
	private static final QueueLog log = QueueLog.getLogger(JmsConsumer.class);
	private QueueConfig.Consumer config;
	private ReceiveListener receive = null;
	private Connection connection = null;
	private Session session = null;
	private MessageConsumer consumer = null;

	public JmsConsumer() {
		config = QueueConfig.getConsumer();
		conn();
	}

	@Override
	public void setOnReceiveListener(ReceiveListener receive) {
		this.receive = receive;
	}

	public synchronized void conn() {
		log.info("get consumer connection");
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD,
				config.ServiceUrl + "?wireFormat.maxInactivityDuration=3000");
		try {
			connection = connectionFactory.createConnection();
			connection.start();
			session = connection.createSession(Boolean.FALSE,
					Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(
					config.QueueName + "?consumer.prefetchSize=10");
			consumer = session.createConsumer(destination);
		} catch (JMSException e) {
			log.error("get consumer onnection error", e);
		}

	}

	@Override
	public synchronized void close() {
		try {
			if (consumer != null) {
				consumer.close();
			}
			if (session != null) {
				session.close();
			}
			if (connection != null) {
				connection.close();
			}
		} catch (JMSException e) {
			log.error("colse error", e);
		}
	}

	@Override
	public void run() {
		while (true) {
			try {
				if (consumer == null) {
					throw new JMSException("jms consumer is not created");
				}
				TextMessage mconsumer = (TextMessage) consumer
						.receive(config.QueueInterval);
				if (mconsumer != null) {
					if (receive != null) {
						receive.receive(mconsumer.getText());
					}
				}
			} catch (JMSException e) {
				log.error("getMessage error", e);
				close();
				conn();
			} catch (Exception e) {
				log.error("receive error", e);
			}
		}
	}

	public ReceiveListener getReceive() {
		return receive;
	}

}
